package tedu.rocket;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.junit.Test;

import java.nio.charset.StandardCharsets;
import java.sql.SQLOutput;
import java.util.List;
@Slf4j
public class MyConsumer {
    //测试push的集群模式消费
    @Test
    public void consume() throws Exception {
        //1. 消费者对象,连接namesrv 设置消费者分组,设置订阅的主题和消息过滤的条件
        // 创建消费对象 push消费 集群模式
        DefaultMQPushConsumer consumer=new DefaultMQPushConsumer();
        // 设置消费模式 集群 广播
        //consumer.setMessageModel(MessageModel.BROADCASTING);
        //consumer.setMessageModel(MessageModel.CLUSTERING);
        consumer.setNamesrvAddr("localhost:9876");
        consumer.setConsumerGroup("c-group02");
        // 订阅的主题
        consumer.subscribe("topic02","*");
        //2. 创建绑定一个消费的监听器 随时监听 rocketmq是否push过来新的消息 消费逻辑就是在监听器实现的
        MessageListenerConcurrently messageListener=
                new MessageListenerConcurrently() {
                    //底层rocketmq客户端消费的核心方法,每个消息都会调用一次这个方法
                    /**
                     * @param list 消息对象 虽然是list但是每次push list元素只有一个
                     * @param consumeConcurrentlyContext 当前消费者身份信息
                     * @return 消费状态 只有2种 CONSUME_SUCCESS成功 RECONSUME_LATER等会再消费
                     */
                    @Override
                    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                        //测试代码消费逻辑: 解析body byte[] 打印字符串
                        MessageExt messageExt = list.get(0);
                        String msgId = messageExt.getMsgId();
                        byte[] body = messageExt.getBody();
                        String msg=new String(body, StandardCharsets.UTF_8);
                        log.info("消费者接收消息,{}",msg);
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                };
        consumer.setMessageListener(messageListener);
        //3. 开启连接 监听逻辑是异步执行 保证能看到异步执行的消费,当前线程 持续死循环
        consumer.start();
        while(true);
    }
}

